package org.lisy.java.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * 并发容器
 * 
 * @author lisy
 */
public class ConcurrentContainer {

	public static void main(String[] args) {
		
		// 并发 HashMap
		concurrentHashMapOperate();
		
		// 并发 ArrayList
		copyOnWriteArrayListOperate();
		
		// 并发 Set
		copyOnWriteArraySetOperate();
		
		// 并发队列 (基于链表)
		concurrentLinkedQueueOperate();
		
		// 并发队列 (基于双向链表)
		concurrentLinkedDequeOperate();
		
		// 基于跳表的并发 Map
		concurrentSkipListMapOperate();
		
		// 基于跳表的并发 Set
		concurrentSkipListSetOperate();

		// 阻塞队列 (基于数组)
		arrayBlockingQueueOperate();
		
		// 阻塞队列 (基于链表)
		linkedBlockingQueueOperate();
		
		// 阻塞队列 (基于双向链表)
		linkedBlockingDequeOperate();
		
		// 线程安全的优先队列
		priorityBlockingQueueOperate();
		
		// 读写成对的队列
		synchronousQueueOperate();
		
		// 基于链表的数据交换队列
		linkedTransferQueueOperate();
		
		// 延时队列
		delayQueueOperate();
	}

	private static void delayQueueOperate() {
		/**
		 * 可以使放入队列的元素在指定的延时后才被消费者取出，元素需要实现 Delayed 接口
		 */
		DelayQueue<Task> delayQueue = new DelayQueue<Task>();
		long now = System.currentTimeMillis();
		delayQueue.put(new Task(now + 1000));
		delayQueue.put(new Task(now + 2000));
        System.out.println(delayQueue);
        for(int i = 0; i < 2; i++) {
            try {
				System.out.println(delayQueue.take());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
        }
	}

	private static void linkedTransferQueueOperate() {
		/**
		 * 实现了接口 TransferQueue，通过 transfer 方法放入元素时，如果发现有线程在阻塞在取元素，会直接把这个元素给等待线程
		 * 如果没有人等着消费，那么会把这个元素放到队列尾部，并且此方法阻塞直到有人读取这个元素。和 SynchronousQueue 有点像，但比它更强大
		 */
		LinkedTransferQueue<Object> linkedTransferQueue = new LinkedTransferQueue<Object>();
		linkedTransferQueue.add("LinkedTransferQueue");
		System.out.println(linkedTransferQueue);
	}

	private static void synchronousQueueOperate() {
		/**
		 * 一个虚假的队列，因为它实际上没有真正用于存储元素的空间，每个插入操作都必须有对应的取出操作，没取出时无法继续放入
		 * 使用场景举例: Executors.newCachedThreadPool()，创建一个缓存线程池
		 */
		SynchronousQueue<Object> synchronousQueue = new SynchronousQueue<Object>();
		System.out.println(synchronousQueue);
        new Thread(()->{
            try{
                System.out.println("take: " + synchronousQueue.take());
            } catch (InterruptedException e){
                e.printStackTrace();
            }
        }).start();
		try {
			Thread.sleep((long)(Math.random() * 1000));
			System.out.println("put: SynchronousQueue");
			synchronousQueue.put("SynchronousQueue");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private static void priorityBlockingQueueOperate() {
		/**
		 * 构造时可以传入一个比较器，可以看做放进去的元素会被排序，然后读取的时候按顺序消费
		 * 某些低优先级的元素可能长期无法被消费，因为不断有更高优先级的元素进来
		 */
		PriorityBlockingQueue<Object> priorityBlockingQueue = new PriorityBlockingQueue<Object>();
		priorityBlockingQueue.add("PriorityBlockingQueue");
		System.out.println(priorityBlockingQueue);
	}

	private static void linkedBlockingDequeOperate() {
		/**
		 * 类似 LinkedBlockingQueue，但提供了双向链表特有的操作
		 */
		LinkedBlockingDeque<Object> linkedBlockingDeque = new LinkedBlockingDeque<Object>();
		linkedBlockingDeque.add("LinkedBlockingDeque");
		System.out.println(linkedBlockingDeque);
	}

	private static void linkedBlockingQueueOperate() {
		/**
		 * 基于链表实现的阻塞队列，想比与不阻塞的 ConcurrentLinkedQueue，它多了一个容量限制，如果不设置默认为 int 最大值
		 */
		LinkedBlockingQueue<Object> linkedBlockingQueue = new LinkedBlockingQueue<Object>();
		linkedBlockingQueue.add("LinkedBlockingQueue");
		System.out.println(linkedBlockingQueue);
	}

	private static void arrayBlockingQueueOperate() {
		/**
		 * 基于数组实现的可阻塞队列，构造时必须制定数组大小，往里面放东西时如果数组满了便会阻塞直到有位置（也支持直接返回和超时等待），通过一个锁 ReentrantLock 保证线程安全
		 * 
		 * java 1.5 后可使用 Condition 替代传统的 Object 的 wait()、notify() 实现线程间的协作，使用 Condition 的 await()、signal() 这种方式实现线程间协作更加安全和高效
		 * 通常来说比较推荐使用 Condition，阻塞队列实际上是使用了 Condition 来模拟线程间协作
		 */
		ArrayBlockingQueue<Object> arrayBlockingQueue = new ArrayBlockingQueue<Object>(5);
		arrayBlockingQueue.add("ArrayBlockingQueue");
		System.out.println(arrayBlockingQueue);
	}

	private static void concurrentSkipListSetOperate() {
		/**
		 * ConcurrentSkipListSet 里面就是一个 ConcurrentSkipListMap
		 */
		ConcurrentSkipListSet<Object> concurrentSkipListSet = new ConcurrentSkipListSet<Object>();
		concurrentSkipListSet.add("ConcurrentSkipListSet");
		System.out.println(concurrentSkipListSet);
	}

	private static void concurrentSkipListMapOperate() {
		/**
		 * SkipList 即跳表，跳表是一种空间换时间的数据结构，通过冗余数据，将链表一层一层索引，达到类似二分查找的效果
		 */
		ConcurrentSkipListMap<Object, Object> concurrentSkipListMap = new ConcurrentSkipListMap<Object, Object>();
		concurrentSkipListMap.put("key", "ConcurrentSkipListMap");
		System.out.println(concurrentSkipListMap);
	}

	private static void concurrentLinkedDequeOperate() {
		/**
		 * 基于双向链表实现的并发队列，可以分别对头尾进行操作
		 * 可先进先出 (FIFO)，也可以先进后出（FILO），当然先进后出的话应该叫它栈了
		 */
		ConcurrentLinkedDeque<Object> concurrentLinkedDeque = new ConcurrentLinkedDeque<Object>();
		concurrentLinkedDeque.add("ConcurrentLinkedDeque");
		System.out.println(concurrentLinkedDeque);
	}

	private static void concurrentLinkedQueueOperate() {
		/**
		 * 基于链表实现的并发队列，使用乐观锁 (CAS) 保证线程安全
		 * 因为数据结构是链表，所以理论上是没有队列大小限制的，也就是说添加数据一定能成功
		 */
		ConcurrentLinkedQueue<Object> concurrentLinkedQueue = new ConcurrentLinkedQueue<Object>();
		concurrentLinkedQueue.add("ConcurrentLinkedQueue");
		System.out.println(concurrentLinkedQueue);
	}

	private static void copyOnWriteArraySetOperate() {
		/**
		 * 基于 CopyOnWriteArrayList 实现（内含一个 CopyOnWriteArrayList 成员变量），也就是说底层是一个数组，每次 add 都要遍历整个集合才能知道是否存在，不存在时需要插入（加锁）
		 * 适用场景：在 CopyOnWriteArrayList 适用场景下加一个，集合别太大（全部遍历）
		 */
		CopyOnWriteArraySet<Object> copyOnWriteArraySet = new CopyOnWriteArraySet<Object>();
		copyOnWriteArraySet.add("CopyOnWriteArraySet");
		System.out.println(copyOnWriteArraySet);
	}

	private static void copyOnWriteArrayListOperate() {
		/**
		 * 底层结构也是数组，当新增和删除元素时会创建一个新的数组，在新的数组中增加或者排除指定对象，最后用新增数组替换原来的数组
		 * 适用场景：由于读操作不加锁，写（增、删、改）操作加锁，因此适用于读多写少的场景
		 * 局限：由于读的时候不会加锁（读的效率高，就和普通 ArrayList 一样），读取的当前副本，因此可能读取到脏数据
		 */
		CopyOnWriteArrayList<Object> copyOnWriteArrayList = new CopyOnWriteArrayList<Object>();
		copyOnWriteArrayList.add("CopyOnWriteArrayList");
		System.out.println(copyOnWriteArrayList);
	}

	private static void concurrentHashMapOperate() {
		/**
		 * JAVA 7 中采用分段锁来减少锁的竞争，JAVA 8 中放弃了分段锁，采用 CAS
		 * 同时为了防止哈希冲突严重时退化成链表，会在链表长度达到阈值（8）后转换成红黑树
		 */
		ConcurrentHashMap<Object, Object> concurrentHashMap = new ConcurrentHashMap<Object, Object>();
		concurrentHashMap.put("key", "ConcurrentHashMap");
		System.out.println(concurrentHashMap);
	}
}

class Task implements Delayed {
	
    long time = System.currentTimeMillis();
    
    public Task(long time) {
        this.time = time;
    }
    
    @Override
    public int compareTo(Delayed o) {
        if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS))
            return -1;
        else if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) 
            return 1;
        else 
            return 0;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(time - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    
    @Override
    public String toString() {
        return "" + time;
    }
}
